package cn.lingyangwl.agile.msg.push.core.channel.websocket;

import cn.lingyangwl.agile.common.security.config.WebSocketSecurityConfig;
import cn.lingyangwl.agile.model.enums.AuthErrorEnum;
import cn.lingyangwl.agile.model.module.auth.LoginUser;
import cn.lingyangwl.agile.model.module.tenant.TenantContext;
import cn.lingyangwl.agile.msg.push.api.model.rqrs.SitePushReq;
import cn.lingyangwl.framework.core.response.Resp;
import cn.lingyangwl.framework.tool.core.CollectionUtils;
import cn.lingyangwl.framework.tool.core.StringUtils;
import cn.lingyangwl.framework.tool.core.exception.BizException;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.stream.Collectors;

/**
 * WebSocket服务类
 * /ws/push/pc/{userId}
 */
@Slf4j
@ServerEndpoint(value = "/ws/push/pc/{userId}", configurator = WebSocketSecurityConfig.class)
@Component
public class PcChannelClient {

    /** 当前在线客户端 数量 */
    private static int onlineClientSize = 0;
    
    private static final Map<String, Set<PcChannelClient>> sessionMap = new ConcurrentHashMap<>();

    /** 与某个客户端的连接会话，需要通过它来给客户端发送数据 */
    private Session session;

    /** 客户端自定义ID */
    private String userId = "";

    /** 用户登录id */
    private String loginId = "";

    /** 租户id */
    private String tenantId = TenantContext.EMPTY_TENANT_ID;

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        try {
            Object obj = session.getUserProperties().get(LoginUser.class.getSimpleName());
            if (Objects.isNull(obj)) {
                log.error("login user info is null");
                throw new BizException(AuthErrorEnum.TOKEN_EXPIRED);
            }
            LoginUser loginUser = (LoginUser) obj;

            //设置当前属性
            this.userId = userId;
            this.loginId = loginUser.getLoginId();
            this.session = session;
            this.tenantId = loginUser.getTenant().getTenantId();

            Set<PcChannelClient> wsServerSet = sessionMap.get(userId);
            if(wsServerSet == null) {
                wsServerSet = new CopyOnWriteArraySet<>();
            }
            wsServerSet.add(this);
            sessionMap.put(this.tenantId + "_" + userId + "_" + loginId, wsServerSet);

            addOnlineCount(); //在线数加1
            log.info("连接开启监听 === userId: {}, loginId: {}, tenantId: {}, 当前在线人数: {}", userId, loginId, tenantId, onlineClientSize);
        } catch (Exception e) {
            log.error("ws监听异常, userId: {}", userId, e);
        }
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("userId: {}, onMessage: {}", this.userId, message);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        Set<PcChannelClient> wsSet = sessionMap.get(this.tenantId + "_" + this.userId + "_" + this.loginId);
        if (Objects.isNull(wsSet)) {
            return;
        }
        wsSet.remove(this);
        if(wsSet.isEmpty()) {
            sessionMap.remove(this.userId);
        }

        subOnlineCount(); //在线数减1
        log.info("连接关闭 === userId: {}, loginId: {}, tenantId: {}, 当前在线人数: {}", userId, this.loginId, tenantId, onlineClientSize);
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("ws发生错误", error);
    }

    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 根据用户唯一标识发送消息,推送消息
     */
    public static void sendMsg(SitePushReq msg) {
        String userId = msg.getUserId();
        String loginId = msg.getLoginId();
        String tenantId = TenantContext.get().getTenantId();
        if (StringUtils.isEmpty(userId)) {
            log.warn("userId is empty, no push");
            throw new BizException("userId 不能为空");
        }
        if (sessionMap.isEmpty()) {
            throw new BizException("无在线用户");
        }

        String msgJson = JSON.toJSONString(Resp.ok(msg));
        try {
            log.info("推送ws消息到浏览器, userId: {}, loginId: {}, msg: {}", userId, loginId, msgJson);

            // 判断是否推送全部, 只针对某一个租户
            if ("*".equals(userId)) {
                // 获取当前租户所有在线用户
                List<PcChannelClient> pushClientList = sessionMap.entrySet().stream()
                        .filter(e -> tenantId.compareTo(e.getKey().split("_")[0]) == 0)
                        .flatMap(e -> e.getValue().stream()).collect(Collectors.toList());

                sendMessageToClients(pushClientList, String.valueOf(tenantId), userId, msgJson);
            } else {
                if ("*".equals(loginId)) {
                    List<PcChannelClient> pushClientList = sessionMap.entrySet().stream()
                            .filter(e -> tenantId.compareTo(e.getKey().split("_")[0]) == 0)
                            .filter(e -> loginId.equals((e.getKey().split("_")[2])))
                            .flatMap(e -> e.getValue().stream()).collect(Collectors.toList());

                    sendMessageToClients(pushClientList, String.valueOf(tenantId), userId, msgJson);
                    return;
                }

                // 发送到某个租户某个用户
                Set<PcChannelClient> clientSet = sessionMap.get(tenantId + "_" + userId + "_" + loginId);
                if (CollectionUtils.isEmpty(clientSet)) {
                    throw new BizException("用户不在线, userId: {}, loginId: {}, tenantId: {}", userId, loginId, tenantId);
                }
                clientSet.forEach(item -> {
                    try {
                        item.sendMessage(msgJson);
                    } catch (Exception e) {
                        log.info("推送消息时异常，userId: {},  loginId: {}, tenantId: {}", item.userId, loginId, item.tenantId, e);
                    }
                });
            }

        } catch (Exception e) {
            log.info("推送消息时异常，userId: {}", userId, e);
            throw new BizException(e.getMessage());
        }
    }

    private static void sendMessageToClients(List<PcChannelClient> pushClientList, String tenantId, String userId, String msgJson) {
        if (CollectionUtils.isEmpty(pushClientList)) {
            throw new BizException("没有任何用户在线, userId: {}, tenantId: {}", userId, tenantId);
        }
        pushClientList.forEach(item -> {
            try {
                item.sendMessage(msgJson);
                log.debug("push msg to pc, tenantId: {}, userId: {}", item.tenantId, item.userId);
            } catch (Exception e) {
                log.info("推送消息时异常, userId: {}, tenantId: {}", item.userId, item.tenantId, e);
            }
        });
    }

    public static synchronized int getOnlineClientSize() {
        return onlineClientSize;
    }

    public static synchronized void addOnlineCount() {
        onlineClientSize++;
    }

    public static synchronized void subOnlineCount() {
        onlineClientSize--;
    }
}
